package com.foreveross.fetl.queue;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import javax.annotation.Resource;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.log4j.Logger;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;

import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.core.domain.task.SourceEntity;
import com.foreveross.fetl.queue.util.ExchangeRateUtil;
import com.foreveross.fetl.queue.util.LogMessageUtils;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.RowkeyUtils;
import com.foreveross.fetl.util.StringUtils;


/**
 * 每小时触发一次
 * 源数据表：  CRAWL_PLANE_FREIGHT_DOUBLE
 *	 抓取日期(年月日时)_渠道_区域ID_起点(使用三字码)_终点(使用三字码)_航班日期(年月日时)_航班ID_5位随即数
 *	         10位 +    5位  + 2位 +    3位   +         3位      +           10位+            8位 +    5位
 *
 *每天的航班历史最低价(即每天抓取多次航班中的最低价格)LP_FLIGHT_HISTORY_DOUBLE
 * rowkey: 渠道(5)_区域ID(2)_起点(使用三字码)(3)_终点(使用三字码)(3)_抓取时间(年月日)(8)
 * _航班日期(年月日)(8)_返程日期（年月日)(8)_航班号(8)_(中转航班号1)(8)_(中转航班号2)(8)
 *
 *
 */
@Component
public class HistoryDoubleDataPreprocess {
	
	private static Logger log = Logger.getLogger("historyDouble");
	
	
	@Resource(name="taskExecutor")
	private  TaskExecutor taskExecutor;
	
	private int tasksNumber;//任务数 多少条rowkey
	
	private int failuresNumber;//失败数
	
	private int threadsNumber;//线程数目
	
	private int executeNumber;//已执行的任务条数
	
	private int processNumber; //process 执行次数
	
	private int insertNumber; //插入条数 
	
	private long excuteStartTime;
	
	private String taskTime=null;
	
	public void  rollBackNumber(){
		tasksNumber=0;
		failuresNumber=0;
		threadsNumber=0;
		executeNumber=0;
		processNumber=0;
		insertNumber=0;
		excuteStartTime = System.currentTimeMillis();
		LogMessageUtils.create("统计往返历史最低价", this.getClass().getSimpleName());
	}
	
	public void execute(){
		Date endTime= new Date();
		rollBackNumber();
		final Calendar c = Calendar.getInstance();
		c.add(Calendar.HOUR_OF_DAY, -1);
		taskTime=DateUtils.format(c.getTime(), "yyyy-MM-dd HH");
		try {
			ExchangeRateUtil.initExchangeRate(DateUtils.format(new Date(), "yyyyMMdd"));
			process(c.getTime(), endTime);
		} catch (Exception e) {
			//this.saveOrUpdateLog(logMessage, 3, StringUtils.getStackTrace(e));
		}
		
	}
	
	public void manualExecute(final String start,String time){
		this.rollBackNumber();
		try {
		List<String> timeList = Arrays.asList(time.split(";"));
		Calendar c = Calendar.getInstance();
		taskTime = start + " " +time;
		for (String s : timeList) {
			String str =start;
			str += s;
			Date startTime = DateUtils.parse(str, "yyyyMMddHH");
			c.setTime(startTime);
			c.add(Calendar.HOUR_OF_DAY, +1);
			Date endTime = c.getTime();
			this.process(startTime, endTime);
		}
		} catch (Exception e) {
			//this.saveOrUpdateLog(logMessage, 3, StringUtils.getStackTrace(e));
			log.error(StringUtils.getStackTrace(e));
		}
}
	
	
	
	public void process(final Date startTime, final Date endTime) {
		processNumber++;
		if(processNumber ==1){
			//重新初始化汇率
			
		}
		SourceEntity source = new SourceEntity();
		List<SourceEntity> listHome=source.findSourceEntityByType(SourceEntity.TYPE_HOME);
		List<SourceEntity> listchannels = source.findSourceEntityByType(SourceEntity.TYPE_CHNNAEL);
		List<String> strchannels= new ArrayList<String>();
		if(null!=listHome){
			for(SourceEntity entity:listHome){
				strchannels.add(entity.getHbaseKey());
			}
		}
		if(null!=listchannels){
			for(SourceEntity entity:listchannels){
				strchannels.add(entity.getHbaseKey());
			}
		}
		Collection<String> channels = strchannels;
		
		log.info("HistoryDoubleDataPreprocess start time：" + DateUtils.format(startTime, "yyyyMMddHH")
				+ "end time：" + DateUtils.format(endTime, "yyyyMMddHH"));
		for (final String channel : channels) {
			
			threadsNumber++;
			
		}
		for (final String channel : channels) {
			
				taskExecutor.execute(new Runnable() {
	
					@Override
					public void run() {
						Calendar cStart = Calendar.getInstance();
						cStart.setTime(startTime);
						Calendar cEnd = Calendar.getInstance();
						cEnd.setTime(endTime);
						while (cStart.before(cEnd)) {
							String startRow=RowkeyUtils.getRowKey(cStart.getTime(), channel,  RowkeyUtils.START_ROW_KEY);
							String endRow=RowkeyUtils.getRowKey(cStart.getTime(), channel,  RowkeyUtils.STOP_ROW_KEY);
							processDataByHour(startRow, endRow);
							cStart.add(Calendar.HOUR_OF_DAY, 1);// 小时数加一
						}
						
					}
					
					
				});
			
		}
	}
	
	private void processDataByHour(String startRowKey,
			String endRowKey) {
		ResultScanner rs = null;
		try {
			rs = HBaseHelper.getRecords("CRAWL_PLANE_FREIGHT_DOUBLE",
					startRowKey, endRowKey,new String[] {"planeinfo","returnPlaneInfosInfo","transitsInfo"},null);
		} catch (Exception e) {
			log.info(StringUtils.getStackTrace(e));
		}
		
		try{
			for (Result res : rs) {
				try {
					tasksNumber++;
					Map<String, Map<String, String>> currentMap = HBaseHelper
							.transformHBaseRowToMap(res);
					Map<String, String> curFlightMap = currentMap.get("planeinfo");
					float flightPrice = Float.parseFloat(curFlightMap
							.get("lowerPrice") != null ? curFlightMap
									.get("lowerPrice") : "0");
					if (flightPrice <= 0) {
						continue;
					}
//					String channelKey=curFlightMap.get("grabChannelNum");
//					if("00006".equals(channelKey)){
//						if(PriceUtil.USDrefePrice==0){
//							PriceUtil.USDrefePrice=this.getUSDrefePrice();
//						}
//						flightPrice=PriceUtil.conversionPrice(flightPrice, PriceUtil.USDrefePrice).floatValue();
//						curFlightMap.put("lowerPrice", flightPrice+"");
//						curFlightMap.put("highPrice", PriceUtil.conversionPrice(Double.parseDouble(null==curFlightMap.get("highPrice")?"0":curFlightMap.get("highPrice")),PriceUtil.USDrefePrice)+"");
//						currentMap.put("planeinfo", curFlightMap);
//					}
					if(curFlightMap.get("flightId")==null) return;
					if(curFlightMap.get("flightId").indexOf("|")>0){
						curFlightMap.put("flightId", curFlightMap.get("flightId").substring(0, curFlightMap.get("flightId").indexOf("|")));
					}
					if(curFlightMap.get("flightId").indexOf(",")>0){
						curFlightMap.put("flightId", curFlightMap.get("flightId").substring(0, curFlightMap.get("flightId").indexOf(",")));
					}
					
					compareHistoryFlightSameDay(currentMap, flightPrice);
					
				}catch (Exception e) {
					failuresNumber++;
					log.info("rowkey[" + new String(res.getRow()) + "]数据预处理错误\r\n");
					log.info(StringUtils.getStackTrace(e));
				}
			}
		}finally{
			if(rs != null) rs.close();
				rs = null;
				executeNumber++;
				if(executeNumber==threadsNumber){
					LogMessageUtils.build("统计往返历史最低价", this.getClass().getSimpleName(), 2, null, 
							failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime);
				}
			
		}
	
	}
	
	
	/**
	 * 比较历史的航班信息
	 * 
	 * @param startRowKey
	 * @param endRowKey
	 * @param planeMap
	 *            最新的航班价格的航班信息
	 * @param newFlightLowPrice
	 *            最新的航班价格
	 */
	private void compareHistoryFlightSameDay( Map<String, Map<String, String>> newPlaneMap,
			float newFlightLowPrice) {
		Map<String, String> flightMap=newPlaneMap.get("planeinfo");
		int start = Integer.parseInt(flightMap.get("grabDate").substring(0, 8)
		+ "00");
		int end = Integer.parseInt(flightMap.get("grabDate").substring(0, 10));
		
		Map<String, Map<String, String>> lowestPlaneMap = newPlaneMap;// 暂存航班信息
		float flightLowPrice = newFlightLowPrice;
		for (int i = start; i < end; i++) {
			ResultScanner rs = null;
			try {
				String newStartRowkey=RowkeyUtils.getRowKey(i+"", flightMap.get("grabChannelNum"),
						flightMap.get("fromCity"), flightMap.get("toCity"),
						flightMap.get("flightBatchStartTime").substring(0, 10), flightMap.get("flightId"), RowkeyUtils.START_ROW_KEY);
				String newStopRowkey=RowkeyUtils.getRowKey(i+"", flightMap.get("grabChannelNum"),
					    flightMap.get("fromCity"), flightMap.get("toCity"),
						flightMap.get("flightBatchStartTime").substring(0, 10), flightMap.get("flightId"), RowkeyUtils.STOP_ROW_KEY);
				
				rs = HBaseHelper.getRecords("CRAWL_PLANE_FREIGHT_DOUBLE",
						newStartRowkey,
						newStopRowkey,new String[] {"planeinfo","returnPlaneInfosInfo","transitsInfo"},null);

				for (Result res : rs) {
					try {// 某一条航班数据的错误，不应该影响其他航班数据的预处理
						Map<String, Map<String, String>> curFlightMap = HBaseHelper
								.transformHBaseRowToMap(res);
						float flightPrice = Float.parseFloat(curFlightMap.get(
								"planeinfo").get("lowerPrice"));
//						String channelKey=curFlightMap.get("planeinfo").get("grabChannelNum");
//						if("00006".equals(channelKey)){
//							if(PriceUtil.USDrefePrice==0){
//								PriceUtil.USDrefePrice=this.getUSDrefePrice();
//							}
//							flightPrice=PriceUtil.conversionPrice(flightPrice, PriceUtil.USDrefePrice).floatValue();
//							curFlightMap.get("planeinfo").put("lowerPrice", flightPrice+"");
//							curFlightMap.get("planeinfo").put("highPrice", PriceUtil.conversionPrice(Double.parseDouble(null==curFlightMap.get("planeinfo").get("highPrice")?"0":curFlightMap.get("planeinfo").get("highPrice")),PriceUtil.USDrefePrice)+"");
//						}
						if (flightPrice < flightLowPrice) {// 比较航班价格 暂存航班信息
							flightLowPrice = flightPrice;
							lowestPlaneMap = curFlightMap;
						}
					} catch (Exception e) {
						log.info("rowkey[" + new String(res.getRow())
								+ "]数据预处理错误\r\n");
						log.info(StringUtils.getStackTrace(e));
					}
				}
			} catch (Exception e1) {
				log.info(StringUtils.getStackTrace(e1));
			}finally{
				if(rs != null) rs.close();
				rs = null;
			}
		}
		saveLowestFlightPrice(lowestPlaneMap);
	}
	
	/**
	 * LP_FLIGHT_HISTORY_DOUBLE rowkey:
	 * 渠道(5)_起点(使用三字码)(3)_终点(使用三字码)(3)_抓取时间
	 * (年月日)(8)_航班日期(年月日)(8)_返程日期（年月日)(8)_航班号(8)_(中转航班号1)(8)_(中转航班号2)(8)
	 * 
	 * @param fcv
	 */
	private void saveLowestFlightPrice(Map<String, Map<String, String>> fcv) {
		Map<String, String> planeMap = fcv.get("planeinfo");
		String grabChannelNum = planeMap.get("grabChannelNum");
		String fromCity = planeMap.get("fromCity");
		String toCity = planeMap.get("toCity");
		//String areaCode = planeMap.get("areaCode");
		String grabDate = planeMap.get("grabDate").substring(0, 8);
		if(planeMap.get("flightId")==null) return;
		if(planeMap.get("flightId").indexOf("|")>0){
			planeMap.put("flightId", planeMap.get("flightId").substring(0, planeMap.get("flightId").indexOf("|")));
		}
		if(planeMap.get("flightId").indexOf(",")>0){
			planeMap.put("flightId", planeMap.get("flightId").substring(0, planeMap.get("flightId").indexOf(",")));
		}
		String flightId = RowkeyUtils.getFlightIdHbaseKey(planeMap
				.get("flightId"));
		String flightBatchStartTime = planeMap.get("flightBatchStartTime")
				.substring(0, 8);
		String flightReturnTime=null;
		boolean isReturn=false;
		for(String infos : fcv.keySet()){
			if(infos.equalsIgnoreCase("returnPlaneInfosInfo")){
				isReturn=true;
				break;
			}
		}
		if(!isReturn){
			return;
		}
		Map<String,String> returnMap=fcv.get("returnPlaneInfosInfo");
		for(String key:returnMap.keySet()){
			if(key.startsWith("flightBatchStartTime")){
				flightReturnTime=returnMap.get(key).substring(0,8);
				break;
			}
			if(key.startsWith("flightBatchDate")){
				flightReturnTime=returnMap.get(key).replace("-", "").substring(0,8);
				break;
			}
		}
		
//		if("00006".equals(grabChannelNum)){
//			//如果是kayak 就按美元汇率转换
//			for(String key:returnMap.keySet()){
//				if(key.startsWith("lowerPrice_")||key.startsWith("highPrice_")){
//					returnMap.put(key, PriceUtil.conversionPrice(Double.parseDouble(null==returnMap.get(key)?"0":returnMap.get(key)),PriceUtil.USDrefePrice)+"");
//				}
//			}
//			fcv.put("returnPlaneInfosInfo", returnMap);
//		}
		
		for(String key:returnMap.keySet()){
			if(key.startsWith("flightBatchStartTime")){
				
			}
		}
		StringBuffer transitFlightNo = new StringBuffer();
	
		if(fcv.containsKey("transitsInfo")){
			Map<String, String> transit =fcv.get("transitsInfo");

			for(String key : transit.keySet()){
				if(key.startsWith("flightNo")){
					String flightNo=transit.get(key);
					if(flightNo.indexOf("|")>0){
						flightNo=flightNo.substring(0, flightNo.indexOf("|"));
					}
					if(flightNo.indexOf(",")>0){
						flightNo=flightNo.substring(0, flightNo.indexOf(","));
					}
					if(flightId.equals(flightNo))
						continue;
					transitFlightNo.append(RowkeyUtils.getFlightIdHbaseKey(flightNo));
				}
			}
		}
		if(grabChannelNum==null || fromCity==null || toCity==null || flightReturnTime==null || grabDate==null){
			return;
		}
		for (int j = transitFlightNo.length(); j < 16; j++) {
			transitFlightNo.append("0");
		}
		//渠道(5)_起点(使用三字码)(3)_终点(使用三字码)(3)_抓取时间(年月日)(8)
		//_航班日期(年月日)(8)_返程日期（年月日)(8)_航班号(8)_(中转航班号1)(8)_(中转航班号2)(8)
		StringBuffer sb = new StringBuffer();
		sb.append(grabChannelNum);
		//sb.append(areaCode); //新rowkey 不要区域
		sb.append(fromCity);
		sb.append(toCity);
		sb.append(grabDate);
		sb.append(flightBatchStartTime);
		sb.append(flightReturnTime);
		sb.append(flightId);
		sb.append(transitFlightNo);
		this.deleteTableRow("LP_FLIGHT_HISTORY_DOUBLE",sb.toString());
		for (Entry<String, Map<String, String>> fe : fcv.entrySet()) {
			Put put = new Put(sb.toString().getBytes());
			for (Entry<String, String> ce : fe.getValue().entrySet()) {
				put.add(fe.getKey().getBytes(), ce.getKey().getBytes(), ce
						.getValue().getBytes());
			}
			try {
				HBaseHelper.addRecord("LP_FLIGHT_HISTORY_DOUBLE", put);
			} catch (RetriesExhaustedWithDetailsException e) {
				failuresNumber++;
				log.error(StringUtils.getStackTrace(e));
				/*try {
					log.info("LP_FLIGHT_HISTORY_DOUBLE"+ sb.toString()+
							" HBase 可能正在进行分区操作，过2分钟再试！"
							+ StringUtils.getStackTrace(e));
					Thread.currentThread().sleep(120 * 1000);
				} catch (InterruptedException e1) {
					e1.printStackTrace();
				}*/
			} catch (Exception e) {
				failuresNumber++;
				log.info(StringUtils.getStackTrace(e));
			}
		}
		insertNumber++;
		
	}
	
	private void deleteTableRow(String tableName, String rowkey){
		try {
			HBaseHelper.delRecord(tableName, rowkey);
		} catch (Exception e2) {
			log.error("insert before delete rowkey="+rowkey+"  "+e2.getMessage());
		}
	}

}
